package config_test

import (
	"context"
	"os"
	"testing"

	"github.com/envoyproxy/ratelimit/test/common"

	pb_struct "github.com/envoyproxy/go-control-plane/envoy/extensions/common/ratelimit/v3"
	pb "github.com/envoyproxy/go-control-plane/envoy/service/ratelimit/v3"
	pb_type "github.com/envoyproxy/go-control-plane/envoy/type/v3"
	stats "github.com/lyft/gostats"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/envoyproxy/ratelimit/src/config"
	mockstats "github.com/envoyproxy/ratelimit/test/mocks/stats"
)

func loadFile(path string) []config.RateLimitConfigToLoad {
	contents, err := os.ReadFile(path)
	if err != nil {
		panic(err)
	}
	configYaml := config.ConfigFileContentToYaml(path, string(contents))
	return []config.RateLimitConfigToLoad{{Name: path, ConfigYaml: configYaml}}
}

func TestBasicConfig(t *testing.T) {
	assert := assert.New(t)
	stats := stats.NewStore(stats.NewNullSink(), false)
	rlConfig := config.NewRateLimitConfigImpl(loadFile("basic_config.yaml"), mockstats.NewMockStatManager(stats), false)
	rlConfig.Dump()
	assert.Equal(rlConfig.IsEmptyDomains(), false)
	assert.EqualValues(0, stats.NewCounter("foo_domain.domain_not_found").Value())
	assert.Nil(rlConfig.GetLimit(context.TODO(), "foo_domain", &pb_struct.RateLimitDescriptor{}))
	assert.EqualValues(1, stats.NewCounter("foo_domain.domain_not_found").Value())
	assert.Nil(rlConfig.GetLimit(context.TODO(), "test-domain", &pb_struct.RateLimitDescriptor{}))
	assert.EqualValues(0, stats.NewCounter("test-domain.domain_not_found").Value())

	rl := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "something"}},
		})
	assert.Nil(rl)

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}},
		})
	assert.Nil(rl)

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key2", Value: "value2"}, {Key: "subkey", Value: "subvalue"}},
		})
	assert.Nil(rl)

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key5", Value: "value5"}, {Key: "subkey5", Value: "subvalue"}},
		})
	assert.Nil(rl)

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}, {Key: "subkey1", Value: "something"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(5, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_SECOND, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1.within_limit").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}, {Key: "subkey1", Value: "subvalue1"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(10, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_SECOND, rl.Limit.Unit)
	assert.EqualValues(
		1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.total_hits").Value())
	assert.EqualValues(
		1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.over_limit").Value())
	assert.EqualValues(
		1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.near_limit").Value())
	assert.EqualValues(
		1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.within_limit").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key2", Value: "something"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(20, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_MINUTE, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.within_limit").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key2", Value: "value2"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(30, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_MINUTE, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key2_value2.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2_value2.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2_value2.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2_value2.within_limit").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key2", Value: "value3"}},
		})
	assert.Nil(rl)

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key3", Value: "foo"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(1, rl.Limit.RequestsPerUnit)
	assert.Empty(rl.Limit.Name, "No name provided in config")
	assert.Equal(pb.RateLimitResponse_RateLimit_HOUR, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key3.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key3.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key3.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key3.within_limit").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key4", Value: "foo"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(1, rl.Limit.RequestsPerUnit)
	assert.EqualValues("key4_rate_limit", rl.Limit.Name, "Name provided in config")
	assert.Equal(pb.RateLimitResponse_RateLimit_DAY, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key4.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key4.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key4.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key4.within_limit").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key6", Value: "foo"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.True(rl.Unlimited)
	assert.EqualValues(1, stats.NewCounter("test-domain.key6.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key6.within_limit").Value())

	// A value for the key with detailed_metric: true
	// should also generate a stat with the value included
	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key7", Value: "unspecified_value"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(70, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_MINUTE, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_unspecified_value.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_unspecified_value.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_unspecified_value.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_unspecified_value.within_limit").Value())

	// Another value for the key with detailed_metric: true
	// should also generate a stat with the value included
	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key7", Value: "another_value"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(70, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_MINUTE, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_another_value.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_another_value.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_another_value.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key7_another_value.within_limit").Value())
}

func TestDomainMerge(t *testing.T) {
	assert := assert.New(t)
	stats := stats.NewStore(stats.NewNullSink(), false)

	files := loadFile("merge_domain_key1.yaml")
	files = append(files, loadFile("merge_domain_key2.yaml")...)

	rlConfig := config.NewRateLimitConfigImpl(files, mockstats.NewMockStatManager(stats), true)
	rlConfig.Dump()
	assert.Nil(rlConfig.GetLimit(context.TODO(), "foo_domain", &pb_struct.RateLimitDescriptor{}))
	assert.Nil(rlConfig.GetLimit(context.TODO(), "test-domain", &pb_struct.RateLimitDescriptor{}))

	rl := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}},
		})
	assert.NotNil(rl)
	assert.EqualValues(10, rl.Limit.RequestsPerUnit)

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key2", Value: "value2"}},
		})
	assert.NotNil(rl)
	assert.EqualValues(20, rl.Limit.RequestsPerUnit)
}

func TestConfigLimitOverride(t *testing.T) {
	assert := assert.New(t)
	stats := stats.NewStore(stats.NewNullSink(), false)
	rlConfig := config.NewRateLimitConfigImpl(loadFile("basic_config.yaml"), mockstats.NewMockStatManager(stats), false)
	rlConfig.Dump()
	// No matching domain
	assert.Nil(rlConfig.GetLimit(context.TODO(), "foo_domain", &pb_struct.RateLimitDescriptor{
		Limit: &pb_struct.RateLimitDescriptor_RateLimitOverride{
			RequestsPerUnit: 10, Unit: pb_type.RateLimitUnit_DAY,
		},
	}))
	rl := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}, {Key: "subkey1", Value: "something"}},
			Limit: &pb_struct.RateLimitDescriptor_RateLimitOverride{
				RequestsPerUnit: 10, Unit: pb_type.RateLimitUnit_DAY,
			},
		})
	assert.Equal("test-domain.key1_value1.subkey1_something", rl.FullKey)
	common.AssertProtoEqual(assert, &pb.RateLimitResponse_RateLimit{
		RequestsPerUnit: 10,
		Unit:            pb.RateLimitResponse_RateLimit_DAY,
	}, rl.Limit)
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something.within_limit").Value())

	// Change in override value doesn't erase stats
	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}, {Key: "subkey1", Value: "something"}},
			Limit: &pb_struct.RateLimitDescriptor_RateLimitOverride{
				RequestsPerUnit: 42, Unit: pb_type.RateLimitUnit_HOUR,
			},
		})
	assert.Equal("test-domain.key1_value1.subkey1_something", rl.FullKey)
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	common.AssertProtoEqual(assert, &pb.RateLimitResponse_RateLimit{
		RequestsPerUnit: 42,
		Unit:            pb.RateLimitResponse_RateLimit_HOUR,
	}, rl.Limit)
	assert.EqualValues(2, stats.NewCounter("test-domain.key1_value1.subkey1_something.total_hits").Value())
	assert.EqualValues(2, stats.NewCounter("test-domain.key1_value1.subkey1_something.over_limit").Value())
	assert.EqualValues(2, stats.NewCounter("test-domain.key1_value1.subkey1_something.near_limit").Value())
	assert.EqualValues(2, stats.NewCounter("test-domain.key1_value1.subkey1_something.within_limit").Value())

	// Different value creates a different counter
	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}, {Key: "subkey1", Value: "something_else"}},
			Limit: &pb_struct.RateLimitDescriptor_RateLimitOverride{
				RequestsPerUnit: 42, Unit: pb_type.RateLimitUnit_HOUR,
			},
		})
	assert.Equal("test-domain.key1_value1.subkey1_something_else", rl.FullKey)
	common.AssertProtoEqual(assert, &pb.RateLimitResponse_RateLimit{
		RequestsPerUnit: 42,
		Unit:            pb.RateLimitResponse_RateLimit_HOUR,
	}, rl.Limit)
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.WithinLimit.Inc()
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something_else.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something_else.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something_else.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_something_else.within_limit").Value())
}

func expectConfigPanic(t *testing.T, call func(), expectedError string) {
	assert := assert.New(t)
	defer func() {
		e := recover()
		assert.NotNil(e)
		assert.Equal(expectedError, e.(error).Error())
	}()

	call()
}

func TestEmptyDomain(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("empty_domain.yaml"), mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"empty_domain.yaml: config file cannot have empty domain")
}

func TestDuplicateDomain(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			files := loadFile("basic_config.yaml")
			files = append(files, loadFile("duplicate_domain.yaml")...)
			config.NewRateLimitConfigImpl(files, mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"duplicate_domain.yaml: duplicate domain 'test-domain' in config file")
}

func TestEmptyKey(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("empty_key.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"empty_key.yaml: descriptor has empty key")
}

func TestDuplicateKey(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("duplicate_key.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"duplicate_key.yaml: duplicate descriptor composite key 'test-domain.key1_value1'")
}

func TestDuplicateKeyDomainMerge(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			files := loadFile("merge_domain_key1.yaml")
			files = append(files, loadFile("merge_domain_key1.yaml")...)
			config.NewRateLimitConfigImpl(
				files,
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), true)
		},
		"merge_domain_key1.yaml: duplicate descriptor composite key 'test-domain.key1_value1'")
}

func TestBadLimitUnit(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("bad_limit_unit.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"bad_limit_unit.yaml: invalid rate limit unit 'foo'")
}

func TestReplacesSelf(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("replaces_self.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"replaces_self.yaml: replaces should not contain name of same descriptor")
}

func TestReplacesEmpty(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("replaces_empty.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"replaces_empty.yaml: should not have an empty replaces entry")
}

func TestBadYaml(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("bad_yaml.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"bad_yaml.yaml: error loading config file: yaml: line 2: found unexpected end of stream")
}

func TestMisspelledKey(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("misspelled_key.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"misspelled_key.yaml: config error, unknown key 'ratelimit'")

	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("misspelled_key2.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"misspelled_key2.yaml: config error, unknown key 'requestsperunit'")
}

func TestNonStringKey(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("non_string_key.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"non_string_key.yaml: config error, key is not of type string: 0.25")
}

func TestNonMapList(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("non_map_list.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"non_map_list.yaml: config error, yaml file contains list of type other than map: a")
}

func TestUnlimitedWithRateLimitUnit(t *testing.T) {
	expectConfigPanic(
		t,
		func() {
			config.NewRateLimitConfigImpl(
				loadFile("unlimited_with_unit.yaml"),
				mockstats.NewMockStatManager(stats.NewStore(stats.NewNullSink(), false)), false)
		},
		"unlimited_with_unit.yaml: should not specify rate limit unit when unlimited")
}

func TestShadowModeConfig(t *testing.T) {
	assert := assert.New(t)
	stats := stats.NewStore(stats.NewNullSink(), false)

	rlConfig := config.NewRateLimitConfigImpl(loadFile("shadowmode_config.yaml"), mockstats.NewMockStatManager(stats), false)
	rlConfig.Dump()

	rl := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}, {Key: "subkey1", Value: "something"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()

	assert.Equal(rl.ShadowMode, false)
	assert.EqualValues(5, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_SECOND, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1.near_limit").Value())
	assert.EqualValues(0, stats.NewCounter("test-domain.key1_value1.subkey1.shadow_mode").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key1", Value: "value1"}, {Key: "subkey1", Value: "subvalue1"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.ShadowMode.Inc()

	assert.Equal(rl.ShadowMode, true)
	assert.EqualValues(10, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_SECOND, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key1_value1.subkey1_subvalue1.shadow_mode").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key2", Value: "something"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	rl.Stats.ShadowMode.Inc()
	assert.Equal(rl.ShadowMode, true)
	assert.EqualValues(20, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_MINUTE, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.near_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2.shadow_mode").Value())

	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "key2", Value: "value2"}},
		})
	rl.Stats.TotalHits.Inc()
	rl.Stats.OverLimit.Inc()
	rl.Stats.NearLimit.Inc()
	assert.Equal(rl.ShadowMode, false)
	assert.EqualValues(30, rl.Limit.RequestsPerUnit)
	assert.Equal(pb.RateLimitResponse_RateLimit_MINUTE, rl.Limit.Unit)
	assert.EqualValues(1, stats.NewCounter("test-domain.key2_value2.total_hits").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2_value2.over_limit").Value())
	assert.EqualValues(1, stats.NewCounter("test-domain.key2_value2.near_limit").Value())
	assert.EqualValues(0, stats.NewCounter("test-domain.key2_value2.shadow_mode").Value())
}

func TestWildcardConfig(t *testing.T) {
	assert := assert.New(t)
	stats := stats.NewStore(stats.NewNullSink(), false)
	rlConfig := config.NewRateLimitConfigImpl(loadFile("wildcard.yaml"), mockstats.NewMockStatManager(stats), false)
	rlConfig.Dump()

	// Baseline to show wildcard works like no value
	withoutVal1 := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "noVal", Value: "foo1"}},
		})
	withoutVal2 := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "noVal", Value: "foo2"}},
		})
	assert.NotNil(withoutVal1)
	assert.Equal(withoutVal1, withoutVal2)
	// Verify stats keys for no value descriptors
	assert.Equal("test-domain.noVal", withoutVal1.Stats.Key, "No value descriptor should use just the key")
	assert.Equal(withoutVal1.Stats.Key, withoutVal1.FullKey, "FullKey should match Stats.Key")
	assert.Equal(withoutVal1.Stats.Key, withoutVal2.Stats.Key, "All no-value matches should have same stats key")

	// Matches multiple wildcard values and results are equal
	wildcard1 := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "wild", Value: "foo1"}},
		})
	wildcard2 := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "wild", Value: "foo2"}},
		})
	wildcard3 := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "nestedWild", Value: "val1"}, {Key: "wild", Value: "goo2"}},
		})
	assert.NotNil(wildcard1)
	assert.Equal(wildcard1, wildcard2)
	assert.Equal("test-domain.wild_foo*", wildcard1.Stats.Key, "Wildcard stats key should include the wildcard pattern with *")
	assert.Equal("test-domain.wild_foo*", wildcard1.FullKey, "Wildcard FullKey should match Stats.Key")
	assert.Equal("test-domain.wild_foo*", wildcard2.Stats.Key, "All wildcard matches should have same stats key with wildcard pattern")
	assert.NotNil(wildcard3)
	assert.Equal("test-domain.nestedWild_val1.wild_goo*", wildcard3.Stats.Key, "Nested wildcard stats key should include the wildcard pattern with *")
	assert.Equal("test-domain.nestedWild_val1.wild_goo*", wildcard3.FullKey, "Nested wildcard FullKey should match Stats.Key")

	// Doesn't match non-matching values
	noMatch := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "wild", Value: "bar"}},
		})
	assert.Nil(noMatch)

	// Non-wildcard values don't eager match
	eager := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "noWild", Value: "foo1"}},
		})
	assert.Nil(eager)

	// Wildcard in the middle of value is not supported.
	midWildcard := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "midWildcard", Value: "barab"}},
		})
	assert.Nil(midWildcard)
}

func TestDetailedMetric(t *testing.T) {
	assert := require.New(t)
	stats := stats.NewStore(stats.NewNullSink(), false)

	// Descriptor config with a realistic nested setup, that is re-used across
	// multiple test cases.
	realisticExampleConfig := &config.YamlRoot{
		Domain: "nested",
		Descriptors: []config.YamlDescriptor{
			{
				Key: "key_1",
				RateLimit: &config.YamlRateLimit{
					RequestsPerUnit: 500,
					Unit:            "minute",
				},
				DetailedMetric: true,
			},
			{
				Key:   "key_1",
				Value: "some-value-for-key-1",
				RateLimit: &config.YamlRateLimit{
					RequestsPerUnit: 500,
					Unit:            "minute",
				},
			},
			{
				Key: "key_2",
				RateLimit: &config.YamlRateLimit{
					RequestsPerUnit: 5000,
					Unit:            "minute",
				},
				Descriptors: []config.YamlDescriptor{
					{
						Key:            "key_3",
						DetailedMetric: true,
					},
					{
						Key:   "key_3",
						Value: "requested-key3-value",
						RateLimit: &config.YamlRateLimit{
							RequestsPerUnit: 5,
							Unit:            "minute",
						},
						DetailedMetric: true,
					},
				},
				DetailedMetric: true,
			},
			{
				Key:   "key_2",
				Value: "specific-id",
				RateLimit: &config.YamlRateLimit{
					RequestsPerUnit: 50000,
					Unit:            "minute",
				},
				Descriptors: []config.YamlDescriptor{
					{
						Key:            "key_3",
						DetailedMetric: true,
					},
					{
						Key:   "key_3",
						Value: "requested-key3-value",
						RateLimit: &config.YamlRateLimit{
							RequestsPerUnit: 100,
							Unit:            "minute",
						},
						DetailedMetric: true,
					},
				},
				DetailedMetric: true,
			},
		},
	}

	tests := []struct {
		name             string
		config           []config.RateLimitConfigToLoad
		request          *pb_struct.RateLimitDescriptor
		expectedStatsKey string
	}{
		{
			name: "nested with no values but request only top-level key",
			config: []config.RateLimitConfigToLoad{
				{
					ConfigYaml: &config.YamlRoot{
						Domain: "nested",
						Descriptors: []config.YamlDescriptor{
							{
								Key: "key-1",
								RateLimit: &config.YamlRateLimit{
									RequestsPerUnit: 100,
									Unit:            "minute",
								},
								Descriptors: []config.YamlDescriptor{
									{
										Key: "key-2",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 5,
											Unit:            "minute",
										},
									},
								},
								DetailedMetric: true,
							},
						},
					},
				},
			},
			request: &pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{
						Key:   "key-1",
						Value: "value-1",
					},
				},
			},
			expectedStatsKey: "nested.key-1_value-1",
		},
		{
			name: "nested with no values but request only top-level key with no detailed metric",
			config: []config.RateLimitConfigToLoad{
				{
					ConfigYaml: &config.YamlRoot{
						Domain: "nested",
						Descriptors: []config.YamlDescriptor{
							{
								Key: "key-1",
								RateLimit: &config.YamlRateLimit{
									RequestsPerUnit: 100,
									Unit:            "minute",
								},
								Descriptors: []config.YamlDescriptor{
									{
										Key: "key-2",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 5,
											Unit:            "minute",
										},
									},
								},
								DetailedMetric: false,
							},
						},
					},
				},
			},
			request: &pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{
						Key:   "key-1",
						Value: "value-1",
					},
				},
			},
			expectedStatsKey: "nested.key-1",
		},
		{
			name: "nested with no values and request fully nested descriptors",
			config: []config.RateLimitConfigToLoad{
				{
					ConfigYaml: &config.YamlRoot{
						Domain: "nested",
						Descriptors: []config.YamlDescriptor{
							{
								Key: "key-1",
								RateLimit: &config.YamlRateLimit{
									RequestsPerUnit: 100,
									Unit:            "minute",
								},
								Descriptors: []config.YamlDescriptor{
									{
										Key: "key-2",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 5,
											Unit:            "minute",
										},
										DetailedMetric: true,
									},
								},
								DetailedMetric: true,
							},
						},
					},
				},
			},
			request: &pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{
						Key:   "key-1",
						Value: "value-1",
					},
					{
						Key:   "key-2",
						Value: "value-2",
					},
				},
			},
			expectedStatsKey: "nested.key-1_value-1.key-2_value-2",
		},
		{
			name: "nested with no values and request fully nested descriptors with no detailed metric",
			config: []config.RateLimitConfigToLoad{
				{
					ConfigYaml: &config.YamlRoot{
						Domain: "nested",
						Descriptors: []config.YamlDescriptor{
							{
								Key: "key-1",
								RateLimit: &config.YamlRateLimit{
									RequestsPerUnit: 100,
									Unit:            "minute",
								},
								Descriptors: []config.YamlDescriptor{
									{
										Key: "key-2",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 5,
											Unit:            "minute",
										},
										DetailedMetric: false,
									},
								},
								DetailedMetric: false,
							},
						},
					},
				},
			},
			request: &pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{
						Key:   "key-1",
						Value: "value-1",
					},
					{
						Key:   "key-2",
						Value: "value-2",
					},
				},
			},
			expectedStatsKey: "nested.key-1.key-2",
		},
		{
			name: "test nested descriptors with simple request",
			config: []config.RateLimitConfigToLoad{
				{
					ConfigYaml: realisticExampleConfig,
				},
			},
			request: &pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{
						Key:   "key_1",
						Value: "value-for-key-1",
					},
				},
			},
			expectedStatsKey: "nested.key_1_value-for-key-1",
		},
		{
			name: "test nested only second descriptor request not nested",
			config: []config.RateLimitConfigToLoad{
				{
					ConfigYaml: realisticExampleConfig,
				},
			},
			request: &pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{
						Key:   "key_2",
						Value: "key-2-value",
					},
				},
			},
			expectedStatsKey: "nested.key_2_key-2-value",
		},
		{
			name: "test nested descriptors with nested request",
			config: []config.RateLimitConfigToLoad{
				{
					ConfigYaml: realisticExampleConfig,
				},
			},
			request: &pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{
						Key:   "key_2",
						Value: "key-2-value",
					},
					{
						Key:   "key_3",
						Value: "requested-key3-value",
					},
				},
			},
			expectedStatsKey: "nested.key_2_key-2-value.key_3_requested-key3-value",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			rlConfig := config.NewRateLimitConfigImpl(tt.config, mockstats.NewMockStatManager(stats), true)
			rlConfig.Dump()

			rl := rlConfig.GetLimit(
				context.TODO(), "nested",
				tt.request,
			)
			assert.NotNil(rl)
			assert.Equal(tt.expectedStatsKey, rl.Stats.Key)
		})
	}
}

func TestValueToMetric_UsesRuntimeValuesInStats(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "domain",
				Descriptors: []config.YamlDescriptor{
					{
						Key:           "route",
						ValueToMetric: true,
						Descriptors: []config.YamlDescriptor{
							{
								Key:           "http_method",
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key: "subject_id",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 60,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)

	rl := rlConfig.GetLimit(
		context.TODO(), "domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "draw"},
				{Key: "http_method", Value: "GET"},
				{Key: "subject_id", Value: "123"},
			},
		},
	)
	asrt.NotNil(rl)

	// Should include actual runtime values for keys that set value_to_metric: true
	expectedKey := "domain.route_draw.http_method_GET.subject_id"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Increment a couple of counters to ensure the key is actually used in stats
	rl.Stats.TotalHits.Inc()
	rl.Stats.WithinLimit.Inc()

	asrt.EqualValues(1, store.NewCounter(expectedKey+".total_hits").Value())
	asrt.EqualValues(1, store.NewCounter(expectedKey+".within_limit").Value())
}

func TestValueToMetric_DefaultKeyIncludesValueAtThatLevel(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "d",
				Descriptors: []config.YamlDescriptor{
					{
						Key:           "k1",
						ValueToMetric: true,
						Descriptors: []config.YamlDescriptor{
							{
								Key: "k2",
								RateLimit: &config.YamlRateLimit{
									RequestsPerUnit: 1,
									Unit:            "second",
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
	rl := rlConfig.GetLimit(
		context.TODO(), "d",
		&pb_struct.RateLimitDescriptor{Entries: []*pb_struct.RateLimitDescriptor_Entry{
			{Key: "k1", Value: "A"},
			{Key: "k2", Value: "foo"},
		}},
	)
	asrt.NotNil(rl)
	asrt.Equal("d.k1_A.k2", rl.Stats.Key)
}

func TestValueToMetric_MidLevelOnly(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "d",
				Descriptors: []config.YamlDescriptor{
					{
						Key: "k1",
						Descriptors: []config.YamlDescriptor{
							{
								Key:           "k2",
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key:       "k3",
										RateLimit: &config.YamlRateLimit{RequestsPerUnit: 1, Unit: "second"},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
	rl := rlConfig.GetLimit(
		context.TODO(), "d",
		&pb_struct.RateLimitDescriptor{Entries: []*pb_struct.RateLimitDescriptor_Entry{
			{Key: "k1", Value: "X"},
			{Key: "k2", Value: "Y"},
			{Key: "k3", Value: "Z"},
		}},
	)
	asrt.NotNil(rl)
	// k1 has no flag -> just key; k2 has flag -> include value
	asrt.Equal("d.k1.k2_Y.k3", rl.Stats.Key)
}

func TestValueToMetric_NoFlag_Unchanged(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "d",
				Descriptors: []config.YamlDescriptor{
					{
						Key: "k1",
						Descriptors: []config.YamlDescriptor{
							{
								Key:       "k2",
								RateLimit: &config.YamlRateLimit{RequestsPerUnit: 1, Unit: "second"},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
	rl := rlConfig.GetLimit(
		context.TODO(), "d",
		&pb_struct.RateLimitDescriptor{Entries: []*pb_struct.RateLimitDescriptor_Entry{
			{Key: "k1", Value: "X"},
			{Key: "k2", Value: "Y"},
		}},
	)
	asrt.NotNil(rl)
	// No flags anywhere -> same as old behavior when default matched at k1
	asrt.Equal("d.k1.k2", rl.Stats.Key)
}

func TestValueToMetric_DoesNotOverrideDetailedMetric(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "domain",
				Descriptors: []config.YamlDescriptor{
					{
						Key:           "route",
						ValueToMetric: true,
						Descriptors: []config.YamlDescriptor{
							{
								Key:           "http_method",
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key:            "subject_id",
										DetailedMetric: true,
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 60,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)

	rl := rlConfig.GetLimit(
		context.TODO(), "domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "draw"},
				{Key: "http_method", Value: "GET"},
				{Key: "subject_id", Value: "123"},
			},
		},
	)
	asrt.NotNil(rl)

	// With detailed_metric at the leaf, the detailed metric key should be used, regardless of value_to_metric flags
	expectedKey := "domain.route_draw.http_method_GET.subject_id_123"
	asrt.Equal(expectedKey, rl.Stats.Key)

	rl.Stats.TotalHits.Inc()
	asrt.EqualValues(1, store.NewCounter(expectedKey+".total_hits").Value())
}

func TestValueToMetric_WithConfiguredValues(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "test-domain",
				Descriptors: []config.YamlDescriptor{
					{
						Key:           "route",
						ValueToMetric: true,
						Descriptors: []config.YamlDescriptor{
							{
								Key:           "http_method",
								Value:         "GET", // Configured value in descriptor
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key: "subject_id",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 60,
											Unit:            "minute",
										},
									},
								},
							},
							{
								Key:           "http_method",
								Value:         "POST", // Another configured value
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key: "subject_id",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 30,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)

	// Test GET path - should include runtime value for route, but use configured value for http_method
	rl := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "api"},
				{Key: "http_method", Value: "GET"},
				{Key: "subject_id", Value: "user123"},
			},
		},
	)
	asrt.NotNil(rl)
	asrt.EqualValues(60, rl.Limit.RequestsPerUnit)
	// route has value_to_metric=true, so includes runtime value; http_method has configured value, so uses that
	expectedKey := "test-domain.route_api.http_method_GET.subject_id"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Test POST path - should include runtime value for route, but use configured value for http_method
	rl = rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "api"},
				{Key: "http_method", Value: "POST"},
				{Key: "subject_id", Value: "user456"},
			},
		},
	)
	asrt.NotNil(rl)
	asrt.EqualValues(30, rl.Limit.RequestsPerUnit)
	expectedKey = "test-domain.route_api.http_method_POST.subject_id"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Test that stats are actually created with the correct keys
	rl.Stats.TotalHits.Inc()
	asrt.EqualValues(1, store.NewCounter(expectedKey+".total_hits").Value())
}

func TestValueToMetric_WithWildcard(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "domain",
				Descriptors: []config.YamlDescriptor{
					{
						Key:           "user",
						ValueToMetric: true,
						Descriptors: []config.YamlDescriptor{
							{
								Key:           "action",
								Value:         "read*", // Wildcard pattern
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key: "resource",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 100,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)

	// Test wildcard matching with value_to_metric - should include full runtime value
	rl := rlConfig.GetLimit(
		context.TODO(), "domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "user", Value: "alice"},
				{Key: "action", Value: "readfile"}, // Matches "read*" wildcard
				{Key: "resource", Value: "documents"},
			},
		},
	)
	asrt.NotNil(rl)
	asrt.EqualValues(100, rl.Limit.RequestsPerUnit)
	// Both user and action should include their full runtime values due to value_to_metric
	expectedKey := "domain.user_alice.action_readfile.resource"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Test another wildcard match
	rl = rlConfig.GetLimit(
		context.TODO(), "domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "user", Value: "bob"},
				{Key: "action", Value: "readdata"}, // Also matches "read*" wildcard
				{Key: "resource", Value: "database"},
			},
		},
	)
	asrt.NotNil(rl)
	expectedKey = "domain.user_bob.action_readdata.resource"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Test that stats are actually created with the correct keys
	rl.Stats.TotalHits.Inc()
	asrt.EqualValues(1, store.NewCounter(expectedKey+".total_hits").Value())
}

func TestValueToMetric_WithEmptyValue(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "domain",
				Descriptors: []config.YamlDescriptor{
					{
						Key:           "route",
						ValueToMetric: true,
						Descriptors: []config.YamlDescriptor{
							{
								Key:           "http_method",
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key: "subject_id",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 60,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)

	// Test with empty value for route - should not include underscore and empty value
	rl := rlConfig.GetLimit(
		context.TODO(), "domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: ""}, // Empty value
				{Key: "http_method", Value: "GET"},
				{Key: "subject_id", Value: "123"},
			},
		},
	)
	asrt.NotNil(rl)

	// Should not include underscore and empty value for route
	expectedKey := "domain.route.http_method_GET.subject_id"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Test with empty value for http_method - should not include underscore and empty value
	rl = rlConfig.GetLimit(
		context.TODO(), "domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "draw"},
				{Key: "http_method", Value: ""}, // Empty value
				{Key: "subject_id", Value: "123"},
			},
		},
	)
	asrt.NotNil(rl)

	// Should not include underscore and empty value for http_method
	expectedKey = "domain.route_draw.http_method.subject_id"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Test with empty value for both - should not include underscores and empty values
	rl = rlConfig.GetLimit(
		context.TODO(), "domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: ""},       // Empty value
				{Key: "http_method", Value: ""}, // Empty value
				{Key: "subject_id", Value: "123"},
			},
		},
	)
	asrt.NotNil(rl)

	// Should not include underscores and empty values
	expectedKey = "domain.route.http_method.subject_id"
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Increment counters to ensure the keys are actually used in stats
	rl.Stats.TotalHits.Inc()
	rl.Stats.WithinLimit.Inc()

	asrt.EqualValues(1, store.NewCounter(expectedKey+".total_hits").Value())
	asrt.EqualValues(1, store.NewCounter(expectedKey+".within_limit").Value())
}

// TestValueToMetric_FullKeyMatchesStatsKey verifies that rateLimit.FullKey always matches
// rateLimit.Stats.Key. This is important for debugging and log/metric correlation.
// FullKey is used in debug logs, while Stats.Key is used for actual metrics.
func TestValueToMetric_FullKeyMatchesStatsKey(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	cfg := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "test-domain",
				Descriptors: []config.YamlDescriptor{
					{
						Key:           "route",
						ValueToMetric: true,
						Descriptors: []config.YamlDescriptor{
							{
								Key:           "http_method",
								ValueToMetric: true,
								Descriptors: []config.YamlDescriptor{
									{
										Key: "subject_id",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 60,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)

	// Test case 1: value_to_metric enabled - FullKey should match Stats.Key
	rl := rlConfig.GetLimit(
		context.TODO(), "test-domain",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "api"},
				{Key: "http_method", Value: "GET"},
				{Key: "subject_id", Value: "user123"},
			},
		},
	)
	asrt.NotNil(rl)
	asrt.Equal(rl.FullKey, rl.Stats.Key, "FullKey should match Stats.Key when value_to_metric is enabled")
	expectedKey := "test-domain.route_api.http_method_GET.subject_id"
	asrt.Equal(expectedKey, rl.FullKey)
	asrt.Equal(expectedKey, rl.Stats.Key)

	// Test case 2: value_to_metric disabled - FullKey should match Stats.Key
	cfgNoValueToMetric := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "test-domain-2",
				Descriptors: []config.YamlDescriptor{
					{
						Key: "route",
						Descriptors: []config.YamlDescriptor{
							{
								Key: "http_method",
								Descriptors: []config.YamlDescriptor{
									{
										Key: "subject_id",
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 60,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig2 := config.NewRateLimitConfigImpl(cfgNoValueToMetric, mockstats.NewMockStatManager(store), false)
	rl2 := rlConfig2.GetLimit(
		context.TODO(), "test-domain-2",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "api"},
				{Key: "http_method", Value: "GET"},
				{Key: "subject_id", Value: "user123"},
			},
		},
	)
	asrt.NotNil(rl2)
	asrt.Equal(rl2.FullKey, rl2.Stats.Key, "FullKey should match Stats.Key even when value_to_metric is disabled")

	// Test case 3: detailed_metric enabled - FullKey should match Stats.Key
	cfgDetailedMetric := []config.RateLimitConfigToLoad{
		{
			Name: "inline",
			ConfigYaml: &config.YamlRoot{
				Domain: "test-domain-3",
				Descriptors: []config.YamlDescriptor{
					{
						Key: "route",
						Descriptors: []config.YamlDescriptor{
							{
								Key: "http_method",
								Descriptors: []config.YamlDescriptor{
									{
										Key:            "subject_id",
										DetailedMetric: true,
										RateLimit: &config.YamlRateLimit{
											RequestsPerUnit: 60,
											Unit:            "minute",
										},
									},
								},
							},
						},
					},
				},
			},
		},
	}

	rlConfig3 := config.NewRateLimitConfigImpl(cfgDetailedMetric, mockstats.NewMockStatManager(store), false)
	rl3 := rlConfig3.GetLimit(
		context.TODO(), "test-domain-3",
		&pb_struct.RateLimitDescriptor{
			Entries: []*pb_struct.RateLimitDescriptor_Entry{
				{Key: "route", Value: "api"},
				{Key: "http_method", Value: "GET"},
				{Key: "subject_id", Value: "user123"},
			},
		},
	)
	asrt.NotNil(rl3)
	asrt.Equal(rl3.FullKey, rl3.Stats.Key, "FullKey should match Stats.Key when detailed_metric is enabled")
}

// TestShareThreshold tests config (ShareThresholdKeyPattern) and metrics (Stats.Key)
// Cache key generation is tested in base_limiter_test.go
func TestShareThreshold(t *testing.T) {
	asrt := assert.New(t)
	stats := stats.NewStore(stats.NewNullSink(), false)
	rlConfig := config.NewRateLimitConfigImpl(loadFile("share_threshold.yaml"), mockstats.NewMockStatManager(stats), false)

	// Test Case 1: Basic share_threshold functionality
	t.Run("Basic share_threshold", func(t *testing.T) {
		testValues := []string{"files/a.pdf", "files/b.csv", "files/c.txt"}
		var rateLimits []*config.RateLimit

		for _, value := range testValues {
			rl := rlConfig.GetLimit(
				context.TODO(), "test-domain",
				&pb_struct.RateLimitDescriptor{
					Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "files", Value: value}},
				})
			asrt.NotNil(rl)
			// Verify config: ShareThresholdKeyPattern is set correctly
			asrt.Equal("files/*", rl.ShareThresholdKeyPattern[0])
			asrt.EqualValues(100, rl.Limit.RequestsPerUnit)
			// Verify stats: All should have the same stats key with wildcard pattern
			// share_threshold: stats key includes wildcard pattern with *
			asrt.Equal("test-domain.files_files/*", rl.Stats.Key)
			rateLimits = append(rateLimits, rl)
		}

		// Verify all have same stats key
		for i := 1; i < len(rateLimits); i++ {
			asrt.Equal(rateLimits[0].Stats.Key, rateLimits[i].Stats.Key, "All values should have same stats key")
			asrt.Equal(rateLimits[i].Stats.Key, rateLimits[i].FullKey, "FullKey should match Stats.Key for all rate limits")
		}
	})

	// Test Case 2: share_threshold with metrics (value_to_metric and detailed_metric)
	// Tests that metrics correctly include wildcard prefix when share_threshold is enabled
	t.Run("share_threshold with metrics", func(t *testing.T) {
		// Test value_to_metric with share_threshold
		// share_threshold takes priority: should use wildcard pattern with *
		rl1 := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "route", Value: "api/v1"},
					{Key: "method", Value: "GET"},
				},
			})
		asrt.NotNil(rl1)
		asrt.Equal("api/*", rl1.ShareThresholdKeyPattern[0])
		asrt.Equal("test-domain.route_api/*.method", rl1.Stats.Key, "share_threshold takes priority over value_to_metric, should use wildcard pattern with *")
		asrt.Equal(rl1.Stats.Key, rl1.FullKey, "FullKey should match Stats.Key")

		// Test detailed_metric with share_threshold
		// share_threshold takes priority: should use wildcard pattern with *
		rl2 := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "service", Value: "svc/user"},
					{Key: "endpoint", Value: "get"},
				},
			})
		asrt.NotNil(rl2)
		asrt.Equal("svc/*", rl2.ShareThresholdKeyPattern[0])
		asrt.True(rl2.DetailedMetric)
		asrt.Equal("test-domain.service_svc/*.endpoint_get", rl2.Stats.Key, "share_threshold takes priority over detailed_metric, should use wildcard pattern with *")
		asrt.Equal(rl2.Stats.Key, rl2.FullKey, "FullKey should match Stats.Key")
	})

	// Test Case 3: Nested wildcards with share_threshold
	t.Run("Nested wildcards", func(t *testing.T) {
		// Nested share_threshold
		rl1 := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "nested", Value: "parent"},
					{Key: "files", Value: "nested/file1.txt"},
				},
			})
		asrt.NotNil(rl1)
		asrt.Equal("nested/*", rl1.ShareThresholdKeyPattern[1])
		asrt.EqualValues(200, rl1.Limit.RequestsPerUnit)
		// Verify stats key includes wildcard pattern for nested share_threshold
		asrt.Equal("test-domain.nested_parent.files_nested/*", rl1.Stats.Key, "Nested share_threshold should include wildcard pattern with *")
		asrt.Equal(rl1.Stats.Key, rl1.FullKey, "FullKey should match Stats.Key")

		// Multiple wildcards
		rl2 := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "files", Value: "top/file1.txt"},
					{Key: "files", Value: "nested/file1.txt"},
				},
			})
		asrt.NotNil(rl2)
		asrt.Equal("top/*", rl2.ShareThresholdKeyPattern[0])
		asrt.Equal("nested/*", rl2.ShareThresholdKeyPattern[1])
		// Verify stats key includes wildcard patterns for multiple share_threshold entries
		asrt.Equal("test-domain.files_top/*.files_nested/*", rl2.Stats.Key, "Multiple share_threshold entries should include all wildcard patterns")
		asrt.Equal(rl2.Stats.Key, rl2.FullKey, "FullKey should match Stats.Key")

		// Parent has share_threshold, child does not
		rl3 := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "path", Value: "path/foo"},
					{Key: "file", Value: "file/doc1"},
					{Key: "type", Value: "pdf"},
				},
			})
		asrt.NotNil(rl3)
		asrt.Equal("path/*", rl3.ShareThresholdKeyPattern[0])
		asrt.Equal("", rl3.ShareThresholdKeyPattern[1])
		// Verify stats key: parent has share_threshold (wildcard pattern), child is wildcard without share_threshold (preserves original behavior with *)
		// Note: file has wildcard pattern file/* in config but share_threshold: false, so it preserves original wildcard behavior
		asrt.Equal("test-domain.path_path/*.file_file/*.type", rl3.Stats.Key, "Parent with share_threshold uses wildcard pattern, child wildcard without share_threshold preserves original behavior")
		asrt.Equal(rl3.Stats.Key, rl3.FullKey, "FullKey should match Stats.Key")

		// Both parent and child have share_threshold
		rl4 := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "user", Value: "user/alice"},
					{Key: "resource", Value: "res/file1"},
					{Key: "action", Value: "read"},
				},
			})
		asrt.NotNil(rl4)
		asrt.Equal("user/*", rl4.ShareThresholdKeyPattern[0])
		asrt.Equal("res/*", rl4.ShareThresholdKeyPattern[1])
		// Verify stats key includes wildcard patterns for both parent and child with share_threshold
		// Note: action is just a key with value, no wildcard, so it doesn't include the value in stats key (no value_to_metric or detailed_metric)
		asrt.Equal("test-domain.user_user/*.resource_res/*.action", rl4.Stats.Key, "Both parent and child with share_threshold should include wildcard patterns, action key without flags uses just the key")
		asrt.Equal(rl4.Stats.Key, rl4.FullKey, "FullKey should match Stats.Key")
	})

	// Test Case 4: Explicit value takes precedence over wildcard
	// Verify: file_test1 uses isolated threshold, file_test2/test23/test123 share threshold
	t.Run("Explicit value precedence over wildcard", func(t *testing.T) {
		// Test explicit match: file_test1 should use isolated threshold (50/min)
		rlTest1 := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "file", Value: "test1"},
				},
			})
		asrt.NotNil(rlTest1, "Should find rate limit for file=test1")
		asrt.EqualValues(50, rlTest1.Limit.RequestsPerUnit, "test1 should have isolated limit of 50")
		// ShareThresholdKeyPattern should be nil for explicit matches (lazy initialization)
		asrt.Nil(rlTest1.ShareThresholdKeyPattern, "ShareThresholdKeyPattern should be nil for explicit matches (no share_threshold)")
		// Verify Stats: test1 should have its own stats key
		asrt.Equal("test-domain.file_test1", rlTest1.Stats.Key, "test1 should have its own stats key")

		// Test wildcard matches: file_test2, file_test23, file_test123 should share threshold (100/min)
		testWildcardValues := []string{"test2", "test23", "test123"}
		var rateLimits []*config.RateLimit

		for _, value := range testWildcardValues {
			rl := rlConfig.GetLimit(
				context.TODO(), "test-domain",
				&pb_struct.RateLimitDescriptor{
					Entries: []*pb_struct.RateLimitDescriptor_Entry{
						{Key: "file", Value: value},
					},
				})
			asrt.NotNil(rl, "Should find rate limit for file=%s", value)

			// Verify RateLimitConfig: Should have share_threshold pattern
			asrt.NotNil(rl.ShareThresholdKeyPattern, "ShareThresholdKeyPattern should be set for %s", value)
			asrt.Equal(1, len(rl.ShareThresholdKeyPattern), "Should have one pattern")
			asrt.Equal("test*", rl.ShareThresholdKeyPattern[0], "Pattern should be test*")
			asrt.EqualValues(100, rl.Limit.RequestsPerUnit, "Should have shared limit of 100")

			// Verify Stats: All wildcard matches should have the same stats key with wildcard pattern
			asrt.Equal("test-domain.file_test*", rl.Stats.Key, "Stats key should include wildcard pattern with * when share_threshold is true")

			rateLimits = append(rateLimits, rl)
		}

		// Verify all wildcard matches have same stats key
		for i := 1; i < len(rateLimits); i++ {
			asrt.Equal(rateLimits[0].Stats.Key, rateLimits[i].Stats.Key, "All wildcard values should have same stats key")
		}

		// Verify stats counters are shared for wildcard matches
		for _, rl := range rateLimits {
			rl.Stats.TotalHits.Inc()
		}
		counterValue := stats.NewCounter("test-domain.file_test*.total_hits").Value()
		asrt.GreaterOrEqual(counterValue, uint64(len(testWildcardValues)), "All wildcard values should increment the same counter")

		// test1 should have its own counter
		rlTest1.Stats.TotalHits.Inc()
		counterTest1 := stats.NewCounter("test-domain.file_test1.total_hits").Value()
		asrt.GreaterOrEqual(counterTest1, uint64(1), "test1 should increment its own counter")
	})
}

// TestWildcardStatsBehavior verifies the stats key behavior for wildcards under different flag combinations
func TestWildcardStatsBehavior(t *testing.T) {
	asrt := assert.New(t)
	store := stats.NewStore(stats.NewNullSink(), false)

	t.Run("No flags - preserves original behavior with wildcard pattern", func(t *testing.T) {
		cfg := []config.RateLimitConfigToLoad{
			{
				Name: "inline",
				ConfigYaml: &config.YamlRoot{
					Domain: "test-domain",
					Descriptors: []config.YamlDescriptor{
						{
							Key:   "wild",
							Value: "foo*",
							RateLimit: &config.YamlRateLimit{
								RequestsPerUnit: 20,
								Unit:            "minute",
							},
						},
					},
				},
			},
		}

		rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
		rl := rlConfig.GetLimit(context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "wild", Value: "foo1"}},
			})
		asrt.NotNil(rl)
		asrt.Equal("test-domain.wild_foo*", rl.Stats.Key)
		asrt.Equal(rl.Stats.Key, rl.FullKey)
	})

	t.Run("value_to_metric - uses runtime value", func(t *testing.T) {
		cfg := []config.RateLimitConfigToLoad{
			{
				Name: "inline",
				ConfigYaml: &config.YamlRoot{
					Domain: "test-domain",
					Descriptors: []config.YamlDescriptor{
						{
							Key:           "wild",
							Value:         "foo*",
							ValueToMetric: true,
							RateLimit: &config.YamlRateLimit{
								RequestsPerUnit: 20,
								Unit:            "minute",
							},
						},
					},
				},
			},
		}

		rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
		rl := rlConfig.GetLimit(context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "wild", Value: "foo1"}},
			})
		asrt.NotNil(rl)
		asrt.Equal("test-domain.wild_foo1", rl.Stats.Key)
	})

	t.Run("share_threshold - uses wildcard pattern with *", func(t *testing.T) {
		cfg := []config.RateLimitConfigToLoad{
			{
				Name: "inline",
				ConfigYaml: &config.YamlRoot{
					Domain: "test-domain",
					Descriptors: []config.YamlDescriptor{
						{
							Key:            "wild",
							Value:          "foo*",
							ShareThreshold: true,
							RateLimit: &config.YamlRateLimit{
								RequestsPerUnit: 20,
								Unit:            "minute",
							},
						},
					},
				},
			},
		}

		rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
		rl := rlConfig.GetLimit(context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "wild", Value: "foo1"}},
			})
		asrt.NotNil(rl)
		asrt.Equal("test-domain.wild_foo*", rl.Stats.Key)
		asrt.Equal(rl.Stats.Key, rl.FullKey)
	})

	t.Run("share_threshold with value_to_metric - share_threshold takes priority", func(t *testing.T) {
		cfg := []config.RateLimitConfigToLoad{
			{
				Name: "inline",
				ConfigYaml: &config.YamlRoot{
					Domain: "test-domain",
					Descriptors: []config.YamlDescriptor{
						{
							Key:            "wild",
							Value:          "foo*",
							ShareThreshold: true,
							ValueToMetric:  true,
							RateLimit: &config.YamlRateLimit{
								RequestsPerUnit: 20,
								Unit:            "minute",
							},
						},
					},
				},
			},
		}

		rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
		rl := rlConfig.GetLimit(context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{{Key: "wild", Value: "foo1"}},
			})
		asrt.NotNil(rl)
		asrt.Equal("test-domain.wild_foo*", rl.Stats.Key)
		asrt.Equal(rl.Stats.Key, rl.FullKey)
	})

	t.Run("value_to_metric in other descriptor - wildcard pattern preserved", func(t *testing.T) {
		cfg := []config.RateLimitConfigToLoad{
			{
				Name: "inline",
				ConfigYaml: &config.YamlRoot{
					Domain: "test-domain",
					Descriptors: []config.YamlDescriptor{
						{
							Key:   "wild",
							Value: "foo*",
							Descriptors: []config.YamlDescriptor{
								{
									Key:           "other",
									ValueToMetric: true,
									Descriptors: []config.YamlDescriptor{
										{
											Key:       "limit",
											RateLimit: &config.YamlRateLimit{RequestsPerUnit: 20, Unit: "minute"},
										},
									},
								},
							},
						},
					},
				},
			},
		}

		rlConfig := config.NewRateLimitConfigImpl(cfg, mockstats.NewMockStatManager(store), false)
		rl := rlConfig.GetLimit(
			context.TODO(), "test-domain",
			&pb_struct.RateLimitDescriptor{
				Entries: []*pb_struct.RateLimitDescriptor_Entry{
					{Key: "wild", Value: "foo1"},
					{Key: "other", Value: "bar"},
					{Key: "limit", Value: "X"},
				},
			},
		)
		asrt.NotNil(rl)
		// Wildcard pattern should be preserved even though value_to_metric is enabled on "other"
		asrt.Equal("test-domain.wild_foo*.other_bar.limit", rl.Stats.Key, "Should preserve wildcard pattern when value_to_metric is only on other descriptor")
		asrt.Equal(rl.Stats.Key, rl.FullKey, "FullKey should match Stats.Key")
	})
}
